package com.zhaosc.rx.example;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import com.google.common.collect.Lists;
import com.zhaosc.rx.bean.User;
import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BooleanSupplier;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.internal.schedulers.SingleScheduler;
import io.reactivex.schedulers.Schedulers;

public class ObserverExample {

	@Test
	public void testObseval() {
		/*
		 * 1.创建Observable被观察者
		 */
		Observable<User> observable = Observable.create(new ObservableOnSubscribe<User>() {

			@Override
			public void subscribe(ObservableEmitter<User> emitter) throws Exception {

				// 当 Observable 要发送普通事件时，就会调用这个方法
				long start = System.currentTimeMillis();
				while (true) {
					emitter.onNext(new User(System.currentTimeMillis(), "zhaosc", "zscpwd"));
					long costTime = System.currentTimeMillis() - start;
					if (costTime >= 1000) {
						// 结束
						emitter.onComplete();
						emitter.onError(new RuntimeException("costTime:" + costTime));
						break;
					}
				}

			}
		});

		/*
		 * 2.创建Observer观察者 3.subscribe()订阅操作
		 */
		observable.subscribe(new Observer<User>() {

			@Override
			public void onSubscribe(Disposable d) {
				if (!d.isDisposed()) {
					System.out.println("******onSubscribe*******");
				}
			}

			@Override
			public void onNext(User user) {

				System.out.println(user);
			}

			@Override
			public void onError(Throwable t) {
				t.printStackTrace();
			}

			@Override
			public void onComplete() {
				System.out.println("******onComplete******");
			}

		});

	}

	/**
	 * 生命周期
	 */
	@Test
	public void doSomething() {
		Observable.just("hello", "world", "javaRX").doOnNext((s) -> {
			System.out.println("onNext is:" + s);
		}).doOnError(e -> new RuntimeException(e.getMessage())).doAfterNext((s) -> {
			System.out.println("doAfterNext is:" + s);
		}).doAfterTerminate(() -> {
			System.out.println("doAfterTerminate");
		}).doOnComplete(() -> {
			System.out.println("doOnComplete");
		}).doOnSubscribe((s) -> {
			System.out.println("doOnSubscribe is:" + s);
		}).doFinally(() -> {
			System.out.println("doFinally");
		}).doOnEach((s) -> {
			System.out.println("doOnEach is:" + s);
		}).doOnLifecycle((s) -> {
			System.out.println("doOnLifecycle is:" + s);
		}, () -> {
			System.out.println("doOnLifecycle");
		}).subscribe((s) -> {
			System.out.println("subscribe is:" + s);
		});

	}

	@Test
	public void testObserver() {
		// 被观察者
		Observable novel = Observable.create(new ObservableOnSubscribe<String>() {
			@Override
			public void subscribe(ObservableEmitter<String> emitter) throws Exception {
				emitter.onNext("连载1");
				emitter.onNext("连载2");
				emitter.onNext("连载3");
				emitter.onComplete();
				emitter.setCancellable(() -> {
					System.out.println("cancelled");
				});
			}
		});

		// 观察者
		Observer<String> reader = new Observer<String>() {
			Disposable mDisposable;

			@Override
			public void onSubscribe(Disposable d) {
				mDisposable = d;
				System.out.println("onSubscribe");
			}

			@Override
			public void onNext(String value) {
				if ("2".equals(value)) {
					mDisposable.dispose();
					return;
				}
				System.err.println("onNext:" + value);
			}

			@Override
			public void onError(Throwable e) {
				System.err.println("onError=" + e.getMessage());
			}

			@Override
			public void onComplete() {
				System.err.println("onComplete()");
			}
		};

		novel.subscribe(reader);// 一行代码搞定
	}

	@Test
	public void testFromArray() {
		Observable.fromArray("1", "2", "3").subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("*******" + t);
			}
		});

	}

	@Test
	public void testFromIterater() {

		AtomicInteger integer=new AtomicInteger(0);
		Observable.fromIterable(Lists.newArrayList("1", "2"))
		//.repeat(3)
		//.repeat()
//		.repeatUntil(new BooleanSupplier() {
//			
//			@Override
//			public boolean getAsBoolean() throws Exception {
//				return integer.incrementAndGet()>6;
//			}
//		})
		.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                return Observable.timer(10000, TimeUnit.MILLISECONDS);
            }
        })
		.subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("*****" + t);
			}
		});

	}

	@Test
	public void testFromFuture() {
		ExecutorService executorService = Executors.newCachedThreadPool();
		Future<String> future = executorService.submit(new Callable<String>() {

			@Override
			public String call() throws Exception {
				SleepUtils.sleep(5000L);
				return "hello";
			}
		});

		Observable.fromFuture(future, 4, TimeUnit.SECONDS).subscribe(new Consumer<String>() {

			@Override
			public void accept(String t) throws Exception {
				System.out.println("****" + t);
			}
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				System.out.println("onerro");
				t.printStackTrace();
			}
		});
	}
	
	/**
	 * defer 操作符会一直等待直到有观察者订阅它，
	 * 然后它使用 Observable 工厂方法生成一个 Observable
	 */
	
	@Test
	public void testDefer() {
		
		AtomicInteger integer=new AtomicInteger(0);
		Observable<Integer> observable=Observable.defer(new Callable<ObservableSource<Integer>>() {

			@Override
			public ObservableSource<Integer> call() throws Exception {
				return Observable.fromArray(integer.getAndIncrement());
			}
		});
		
		observable.subscribe(new Consumer<Integer>() {

			@Override
			public void accept(Integer t) throws Exception {
				System.out.println("*****consumer1****"+t);
			}
		});
		
		observable.subscribe(new Consumer<Integer>() {

			@Override
			public void accept(Integer t) throws Exception {
				System.out.println("*****consumer2****"+t);
			}
		});
		SleepUtils.sleep(5000L);
	}

	@Test
	public void testInterval() {
		Observable.interval(1, TimeUnit.SECONDS).subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println("**********"+t);
			}
		});
		
		SleepUtils.sleep(5000L);

	}
	
	@Test
	public void testTimer() {
		
		Observable.timer(1, TimeUnit.SECONDS, Schedulers.newThread())
		.subscribe(new Consumer<Long>() {

			@Override
			public void accept(Long t) throws Exception {
				System.out.println(Thread.currentThread().getName()+"HHHH"+t);
			}
		});
		
		SleepUtils.sleep(10000L);
	}

}
